package com.gjy.kafka.resolve.config;

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * 生产者通用配置（幂等或事务皆可使用）
 *
 * @author gjy
 * @version 1.0
 * @since 2025-10-13 15:19:21
 */
public class KafkaProducerConfig {

    /**
     * @param transactionalId 若为 null 则返回普通幂等生产者配置；若不为 null 则返回事务生产者配置
     */
    public static Properties buildConfig(String transactionalId) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 1. 彻底确保不丢失
        props.put(ProducerConfig.ACKS_CONFIG, "all");          // 等所有 ISR
        props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
        props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120_000);
        // 2. 幂等（开启事务时会自动打开）
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        // 3. 为了保证顺序，max.in.flight 请求 <= 5
        props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);

        // -- 事务专有配置 --
        if (transactionalId != null) {
            props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionalId);
        }
        return props;
    }

}
